上一篇文章了解了 LangChain 基本的 Chain。那我想要 Chain 更多的東西，或者是自己客製化的東西然後跟 LLM Chain 在一起，又或者是 LangChain 還有什麼功能可以 Chain 進來，接下來就來一一實作！
為了真正理解 LCEL，要先來了解 LCEL 是如何運作的。首先我們要在 class 中建立一個方法是 __or__，那當我們在兩個函數間呼叫運算子 "|" 的話，如：chain = class_a | class_b。那麼運算子 "|" 就會被轉換為 chain = class_a.__or__(class_b)。這樣講有點抽象，來試著用一些簡單的函數來創建我們自己的版本。(基礎版與進階版是對我而言是這樣，不一定適用所有人～)
# 1.建立 Runnable 物件
class Runnable:
def __init__(self, func):
self.func = func
def __or__(self, other):
def chained_func(*args, **kwargs):
return other(self.func(*args, **kwargs))
return Runnable(chained_func)
def __call__(self, *args, **kwargs):
return self.func(*args, **kwargs)
# 2.建立兩個簡單的函數
def add_five(x):
return x + 5
def multiply_by_two(x):
return x * 2
# 3.將函數轉換成 Runnable 物件
runnable_add_five = Runnable(add_five)
runnable_multiply_by_two = Runnable(multiply_by_two)
# 4.使用 __or__ 連接 Runnable 物件
chain_object = runnable_add_five.__or__(runnable_multiply_by_two)
print(chain_object(3)) # 16
# 5.使用 "|" 連接 Runnable 物件
chain_pipe = runnable_add_five | runnable_multiply_by_two
print(chain_pipe(3)) # 16
程式碼結果探討 🧐：
__init__ 是可以將函數轉換成 Runnable 物件 (第三部分)__or__ 當我們在程式碼中使用運算子 "|" 會來調用 __or__
__call__ 是允許 Runnable 可以像函數一樣被調用，沒設置這個的話 __or__ 沒辦法調用 (第四部分)+5 和 *2
__or__ 或者是運算子 "|" 的結果是一模一樣的from abc import ABC, abstractmethod
class CRunnable(ABC):
def __init__(self):
self.next = None
@abstractmethod
def process(self, data):
"""
This method must be implemented by subclasses to define
data processing behavior.
"""
pass
def invoke(self, data):
processed_data = self.process(data)
if self.next is not None:
return self.next.invoke(processed_data)
return processed_data
def __or__(self, other):
return CRunnableSequence(self, other)
class CRunnableSequence(CRunnable):
def __init__(self, first, second):
super().__init__()
self.first = first
self.second = second
def process(self, data):
return data
def invoke(self, data):
first_result = self.first.invoke(data)
return self.second.invoke(first_result)
class AddTen(CRunnable):
def process(self, data):
result = data + 10
print("AddTen: ", result)
return result
class MultiplyByTwo(CRunnable):
def process(self, data):
result = data * 2
print("Multiply by 2: ", result)
return result
class MinusSix(CRunnable):
def process(self, data):
result = data - 6
print("MinusSix: ", result)
return f"Result: {result}"
a = AddTen()
b = MultiplyByTwo()
c = MinusSix()
chain = a | b | c
result = chain.invoke(10)
print(result)
# AddTen: 20
# Multiply by 2: 40
# MinusSix: 34
# Result: 34
結果探討 🧐：
前面的實作中，都只有使用到 invoke，但這只是很簡單的一個輸入一個 input，LCEL 還有一些通用的調用方法。
import pprint
from langchain_core.prompts import PromptTemplate
from langchain_ffm import ChatFormosaFoundationModel
llm = ChatFormosaFoundationModel(model="ffm-llama3-70b-chat", temperature=0.01)
template = "{league}在{year}賽季的總冠軍是誰？"
prompt = PromptTemplate(input_variables=["league", "year"], template=template)
# 使用 "|" 將 Prompt 和 Model Chain 起來
chain = prompt | llm
# batch 多筆輸入
pprint.pprint(chain.batch([{"league" : "NBA", "year":"2021"}, {"league" : "MLB", "year":"2020"}]))
程式碼結果探討 🧐：
batch 方式可以一次輸入多筆資料。（請忽略 AI 回答錯誤）from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0)
template = "{league}在{year}賽季的總冠軍是誰？"
prompt = PromptTemplate(input_variables=["league", "year"], template=template)
# 使用 "|" 將 Prompt 和 Model Chain 起來
chain = prompt | llm
# stream
for stream in chain.stream({"league" : "NBA", "year":"2020"}):
print(stream.content, end="|", flush=True)
程式碼結果探討 🧐：
stream 方式可以看到 AI 的生成過程，但有些模型可能不支援 LangChain 的 stream 模式，像是台智雲。但台智雲自己有提供 stream 方式可以使用 -> 台智雲 stream 模式。了解 LCEL 的運作流程，也知道怎麼自定義我們自己的 Chain。昨天說到 LangChain 中將運算子 "|" 當作他們的 Chain，那 Prompt 和 Model 就是 Runnable 的部分。為了讓自定義鏈的創建變得盡可能簡單，LangChain 透過 Runnable 來將不管是自定義的還是 LangChain 中原生的，在使用 LCEL 時可以最大限度地提高靈活性。
這個 Runnable 傳遞什麼參數進去就回傳跟 input 一樣的內容，簡單來說就是什麼行為都沒做。
from langchain_core.runnables import RunnablePassthrough
chain = RunnablePassthrough() | RunnablePassthrough() | RunnablePassthrough()
chain.invoke("hello")
# output: hello
程式碼結果探討 🧐：
這個 Runnable 就如同我們第一個實作的 LCEL 的基礎版的部分。
from langchain_core.runnables import RunnableLambda
def add_five(x):
return x + 5
def multiply_by_two(x):
return x * 2
chain = RunnableLambda(add_five) | RunnablePassthrough() | RunnableLambda(multiply_by_two)
chain.invoke(3)
# output: 16
程式碼結果探討 🧐：
一般的 Runnable 都像 Pipeline 那樣依序執行，就是我昨天提到的 Chain 的順序，這個依序執行也可以稱作 RunnableSequence。那 RuunableParallel 提供平行執行的概念，就是可以將 Runnable 並行的概念，實作一個比較好了解平行執行的概念！我以我昨天 few-shot 與 zero-shot 的作為範例！
from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate
from langchain_ffm import ChatFormosaFoundationModel
llm = ChatFormosaFoundationModel(model="ffm-llama3-70b-chat", temperature=0.01)
# from langchain_openai import ChatOpenAI
# llm = ChatOpenAI(model="gpt-4o", temperature=0)
# from langchain_google_genai import ChatGoogleGenerativeAI
# llm = ChatGoogleGenerativeAI(model="gemini-pro", temperature=0)
# 設定欲修正的詞 -> 現金餘額
text = "現金允額"
# zero-shot 的 prompt
zero_shot_template = """糾正以下文字的錯字 : {input}
修正結果 : """
zero_shot_prompt = PromptTemplate(input_variables=["input"], template=zero_shot_template)
# 使用 "|" 將 Prompt 和 Model Chain 起來
zero_shot_chain = zero_shot_prompt | llm
# few-shot 的 prompt
few_shot_template = """修正以下文字的錯字 : {input}
修正結果 : {answer}"""
few_shot_example_prompt = PromptTemplate(input_variables=["input", "answer"], template=few_shot_template)
# 根據上面 Prompt 的參數設定給 AI 的 few-shot
few_shot_examples = [
{
"input": "通貨紅脹",
"answer": "通貨膨脹"
},
{
"input": "政府有一個獎注的補助費用",
"answer": "政府有一個獎助的補助費用"
},
{
"input": "庫藏骨",
"answer": "庫藏股"
},
]
# 使用 FewShotPromptTemplate 模塊
few_shot_prompt = FewShotPromptTemplate(
examples=few_shot_examples,
example_prompt=few_shot_example_prompt,
suffix="修正以下文字的錯字 : {input}",
input_variables=["input"],
)
# 使用 "|" 將 Prompt 和 Model Chain 起來
few_shot_chain = few_shot_prompt | llm
# 執行完 zero-shot 才會執行 few-shot
zero_shot_response = zero_shot_chain.invoke({"input":text})
print(f"Zero Shot 的結果\n欲修正內容 : {text}\n修正結果 : {zero_shot_response.content}")
few_shot_response = few_shot_chain.invoke({"input":text})
print(f'\nFew Shot 的結果\n欲修正內容 : {text}\n修正結果 : {few_shot_response.content}')
# 平行執行
chain = RunnableParallel({'zero-shot': zero_shot_chain, 'few-shot': few_shot_chain})
chain.invoke(text)
程式碼結果探討 🧐：
如果想要對輸入做一些加工或者對 Runnable 結果做修改，可以使用 assign！
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import PromptTemplate
prompt = PromptTemplate.from_template("{input}")
# 1.針對輸入做加工
chain = RunnablePassthrough.assign(input=lambda x: 'Your name is ' + x['input']) | prompt
print(chain.invoke({"input": "Sean"}))
# output: text='Your name is Sean'
# 2.對 Runnable 結果做修改
def assign_func(input):
return "Lulu"
chain = RunnableParallel({"x": RunnablePassthrough()}).assign(y=RunnableLambda(assign_func))
print(chain.invoke({"x": "Sean"}))
# output: {'x': {'x': 'Sean'}, 'y': 'Lulu'}
def fun1(input: dict):
return input.get("y", "Key not found")
def fun2(upper: str):
return str(upper).upper()
new_chain = RunnableLambda(func1) | RunnableLambda(fun2)
print(new_chain.invoke({"y": "Sean"}))
# output: SEAN
final_chain = chain | new_chain
print(final_chain.invoke({"x": "Sean"}))
# output: LULU
程式碼結果探討 🧐：
今天了解了 LCEL 是如何運作的，也了解幾個常用 Runnables。LCEL 可以讓實作變得很簡單也很易於擴充一些功能，LCEL 的優點不僅限於今天聊到的部分，有興趣的人可以參考 Advantages of LCEL。明天來實作將 LangChain 原生的 Runnable 與自定義的 Runnable 透過 LCEL 做結合，也會在多使用一些除了 invoke 之外的好用函數。
今天真的超趕，好險即時趕出來壓線上傳，下次不敢了...
iThome鐵人賽
看影片追技術